package com.common.rediscaffeine;

import com.common.config.CaffeineEntity;
import com.common.config.RedisCaffeineProperties;
import com.common.config.RedisUtil;
import com.github.benmanes.caffeine.cache.Cache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.support.AbstractValueAdaptingCache;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @description: AbstractValueAdaptingCache 已经实现了 Cache 接口，直接继承此类即可
 * @author：nihongyu
 * @date: 2023/7/18
 */
public class RedisCaffeineCache extends AbstractValueAdaptingCache {

    private Logger logger = LoggerFactory.getLogger(RedisCaffeineCache.class);
    private String cacheName;
    private String cachePrefix;
    private String topic;
    private RedisTemplate<Object, Object> redisTemplate;
    private Map<String, ReentrantLock> keyLockMap = new ConcurrentHashMap();
    private Cache<Object, Object> caffeineCache; //使用 Caffeine 作为本地缓存
    private RedisCaffeineProperties redisCaffeineProperties;
    @Autowired
    private RedisUtil redisUtil;

    /**
     * 默认key超时时间 3600s
     */
    private long defaultExpiration = 3600;


    public RedisCaffeineCache(String cacheName,
                              RedisTemplate<Object, Object> redisTemplate,
                              Cache<Object, Object> caffeineCache,
                              RedisCaffeineProperties redisCaffeineProperties) {
        super(redisCaffeineProperties.getAllowNull());
        this.cacheName = cacheName;
        this.redisTemplate = redisTemplate;
        this.cachePrefix = redisCaffeineProperties.getCachePrefix();
        this.caffeineCache = caffeineCache;
        this.redisCaffeineProperties = redisCaffeineProperties;
        this.topic = redisCaffeineProperties.getTopic();
    }


    @Override
    protected Object lookup(Object key) {
        String redisKey = getRedisKey(key);
        Object value = caffeineCache.getIfPresent(key);
        if (value != null) {
            logger.debug("从本地缓存中获得key, the key is : {}", key);
            return value;
        }
        value = redisTemplate.opsForValue().get(redisKey);

        if (value != null) {
            logger.debug("从redis中获得值，将值放到本地缓存中, the key is : {}", redisKey);
            caffeineCache.put(key, value);
        }
        return value;
    }

    @Override
    public String getName() {
        return this.cacheName;
    }

    @Override
    public Object getNativeCache() {
        return null;
    }

    @Override
    public <T> T get(Object key, Callable<T> valueLoader) {
        Object value = lookup(key);
        if (value != null) {
            return (T) value;
        }
        //key 在 redis 和缓存中均不存在,执行原方法，并更新到缓存
        /**
         * 这里加锁是为了并发访问同一个 key 时，valueLoader 只调用一次 （Cache 接口的解释）
         */
        ReentrantLock lock = null;
        try {
            lock = keyLockMap.get(key);
            if (lock == null) {
                logger.debug("create lock for key : {}", key);
                keyLockMap.putIfAbsent(key.toString(), new ReentrantLock());
                lock = keyLockMap.get(key.toString());
            }
            lock.lock();
            value = lookup(key);
            if (value != null) {
                return (T) value;
            }
            //执行原方法获得value
            value = valueLoader.call();
            // toStoreValue 是父类提供的方法
            Object storeValue = toStoreValue(value);
            put(key, storeValue);
            return (T) value;
        } catch (Exception e) {
            throw new ValueRetrievalException(key, valueLoader, e.getCause());
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void put(Object key, Object value) {
        if (!super.isAllowNullValues() && value == null) {
            this.evict(key);
            return;
        }

        long expire = getExpire();
        logger.debug("put：{},expire:{}", getRedisKey(key), expire);
        redisTemplate.opsForValue().set(getRedisKey(key), toStoreValue(value), expire, TimeUnit.MINUTES);

        //缓存变更时通知其他节点清理本地缓存
        push(new CacheMessage(this.cacheName, key));
    }

    @Override
    public void evict(Object key) {
        // 先清除redis中缓存数据，然后清除caffeine中的缓存，避免短时间内如果先清除caffeine缓存后其他请求会再从redis里加载到caffeine中
        redisTemplate.delete(getRedisKey(key));

        push(new CacheMessage(this.cacheName, key));

        caffeineCache.invalidate(key);
    }

    @Override
    public void clear() {
        // 先清除redis中缓存数据，然后清除caffeine中的缓存，避免短时间内如果先清除caffeine缓存后其他请求会再从redis里加载到caffeine中
        //这里使用 scan 遍历（生产环境一般会禁用 keys 函数）
        Collection<String> keys = redisUtil.scan(this.cacheName.concat(":*"));
        redisUtil.deleteObject(keys);

        push(new CacheMessage(this.cacheName, null));
        caffeineCache.invalidateAll();
    }

    public long getExpire() {
        long expire = defaultExpiration;
        List<CaffeineEntity> caffeineEntityList = redisCaffeineProperties.getCaffeineEntityList();
        Long cacheNameExpire = 0L;
        for (int i = 0; i < caffeineEntityList.size(); i++) {
            if (caffeineEntityList.get(i).getCacheName().equals(this.cacheName)) {
                cacheNameExpire = caffeineEntityList.get(i).getRedisExpire();
                break;
            }
        }
        return cacheNameExpire == null ? expire : cacheNameExpire.longValue();
    }

    private String getRedisKey(Object key) {
        String keyStr = this.cacheName.concat(":").concat(key.toString());
        return StringUtils.isEmpty(this.cachePrefix) ? keyStr : this.cachePrefix.concat(":").concat(keyStr);
    }


    /**
     * @description 清理本地缓存
     */
    public void clearLocal(Object key) {
        logger.debug("clear local cache, the key is : {}", key);
        if (key == null) {
            caffeineCache.invalidateAll();
        } else {
            caffeineCache.invalidate(key);
        }
    }

    /**
     * @description 缓存变更时通知其他节点清理本地缓存
     */
    private void push(CacheMessage message) {
        redisTemplate.convertAndSend(topic, message);
    }

}
